package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdStudyWatchVideo extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdStudyWatchVideo().start(
                10012,
                4,
                Constant.TOPIC_DWD_STUDY_WATCH_VIDEO
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //1.设置状态的失效时间
        //tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));
        //2.从dwd_traffic_app_video主题中读取数据  创建动态表
        tableEnv.executeSql("CREATE TABLE app_video (\n" +
                "  `appVideo` map<string, string>,\n" +
                "  `common` string,\n" +
                "  ts bigint,\n" +
                "  pt as proctime()\n" +
              //  "  et as TO_TIMESTAMP_LTZ(ts, 0),\n" +
               // "  WATERMARK FOR et AS et\n" +
                ") " + SQLUtil.getKafkaDDL(Constant.TOPIC_DWD_TRAFFIC_APP_VIDEO, "dwd_study_watch_video"));


        //3.读取观看视频相关数据
        Table videoView = tableEnv.sqlQuery(
                "select\n" +
                        "    `appVideo`['play_sec'] play_sec ,\n" +
                        "    `appVideo`['position_sec'] `position_sec`,\n" +
                        "    `appVideo`['video_id'] video_id,\n" +
                        "`ts`,\n" +
                        "`pt`\n" +
                        "from\n" +
                        "`app_video`"
        );
        //videoView.execute().print();
        //4. 创建动态表和要写入的主题进行映射
        tableEnv.executeSql(
                "create table " + Constant.TOPIC_DWD_STUDY_WATCH_VIDEO + "(" +
                        "play_sec STRING," +
                        "position_sec string," +
                        "video_id string," +
                        "ts bigint," +
                        "pt TIMESTAMP_LTZ(3)," +
                        "primary key(video_id) not enforced " +
                        ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_STUDY_WATCH_VIDEO));

        videoView.executeInsert(Constant.TOPIC_DWD_STUDY_WATCH_VIDEO);

    }
}
